package com.xs.batch.config;

import com.xs.batch.entity.Order;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.PagingQueryProvider;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.jdbc.core.BeanPropertyRowMapper;

import javax.sql.DataSource;
import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableBatchProcessing
public class OrderBatchConfig {

    @Autowired
    private DataSource dataSource;

    @Bean
    public Job closeOrderJob(JobBuilderFactory jobBuilderFactory, Step closeOrderStep) {
        return jobBuilderFactory.get("closeOrderJob")
                .start(closeOrderStep)
                .build();
    }

    @Bean
    public Step closeOrderStep(StepBuilderFactory stepBuilderFactory) {
        return stepBuilderFactory.get("closeOrderStep")
                .<Order, Order>chunk(1000) // 一次处理1000条. 失败回滚本批次
                .reader(orderReader()) // 读数据
                .processor(orderProcessor()) // 处理数据
                .writer(orderWriter()) // 写数据


                // -------- 允许哪些异常被跳过 --------
                .faultTolerant()    // 容错模式
                .skip(SocketTimeoutException.class)  // 需要跳过的异常
                .skip(ConnectException.class)
                .noSkip(IllegalArgumentException.class) // 指定某种异常绝不能跳过
                .skipLimit(5)                 // 最多跳过5条

                // ----------- 也可以加重试 -----------
                .retry(SocketTimeoutException.class)
                .retry(ConnectException.class)
                .retryLimit(3)

                // ----实现并发（要注意线程安全、顺序也无法保证）-----------
                .taskExecutor(new SimpleAsyncTaskExecutor())  // 关键：加TaskExecutor
                .throttleLimit(5)    // 最大并发线程数，比如5

                .build();
    }

    // 数据读取器，从orders表查找24小时未支付订单
    @Bean
    public JdbcCursorItemReader<Order> orderReader() {
        // 查询24小时未支付订单
        return new JdbcCursorItemReaderBuilder<Order>()
                .dataSource(dataSource)
                .name("orderReader")
                .sql("SELECT * FROM orders WHERE status='UNPAID' AND created_at <= ?")
                .preparedStatementSetter(ps -> 
                    ps.setTimestamp(1, Timestamp.valueOf(LocalDateTime.now().minusHours(24))))
                .rowMapper(new BeanPropertyRowMapper<>(Order.class))
                .build();
    }


   /* @Bean
    public JdbcPagingItemReader<Order> orderPagingReader() throws Exception {
        // 设置 where 部分的参数
        Map<String, Object> parameterValues = new HashMap<>();
        parameterValues.put("createdAt", Timestamp.valueOf(LocalDateTime.now().minusHours(24)));
        parameterValues.put("status", "UNPAID");

        // 配置分页查询
        return new JdbcPagingItemReaderBuilder<Order>()
                .dataSource(dataSource)
                .name("orderPagingReader")
                .queryProvider(queryProvider())
                .parameterValues(parameterValues)
                .pageSize(1000)  // 每页查询1000条，可根据情况调整
                .rowMapper(new BeanPropertyRowMapper<>(Order.class))
                .build();
    }

    @Bean
    public PagingQueryProvider queryProvider() throws Exception {

        SqlPagingQueryProviderFactoryBean queryProvider = new SqlPagingQueryProviderFactoryBean();
        queryProvider.setDataSource(dataSource); // 用于判断数据库类型自动生成分页语句
        queryProvider.setSelectClause("SELECT *");
        queryProvider.setFromClause("FROM orders");
        queryProvider.setWhereClause("WHERE status = :status AND created_at <= :createdAt");
        queryProvider.setSortKey("id"); // 必须有唯一主键或唯一索引字段
        return queryProvider.getObject();
    }*/

    // 数据处理器，对每一条订单进行业务处理
    @Bean
    public ItemProcessor<Order, Order> orderProcessor() {
        return order -> {
            order.setStatus("CLOSED");   // 设置订单状态为已关闭
            order.setProcessed(true);   // 标记已处理
            // 可在此处添加调用短信通知、日志记录等操作
            return order;
        };
    }

    // 数据写入器，将处理后的订单批量update入库
    @Bean
    public ItemWriter<Order> orderWriter() {
        return new JdbcBatchItemWriterBuilder<Order>()
                .dataSource(dataSource)
                .sql("UPDATE orders SET status = :status, processed = :processed WHERE id = :id")
                .beanMapped()
                .build();
    }
}